package com.mqstudy.pulsar.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mqstudy.common.model.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.CompletableFuture;

/**
 * Pulsar生产者
 */
@Slf4j
@Component
public class PulsarProducer {
    
    @Value("${pulsar.service-url}")
    private String serviceUrl;
    
    @Value("${pulsar.tenant}")
    private String tenant;
    
    @Value("${pulsar.namespace}")
    private String namespace;
    
    @Autowired
    private ObjectMapper objectMapper;
    
    private PulsarClient client;
    private Producer<String> producer;
    
    // 主题名称
    public static final String TEST_TOPIC = "test-topic";
    public static final String ORDER_TOPIC = "order-topic";
    public static final String USER_TOPIC = "user-topic";
    public static final String STREAM_TOPIC = "stream-topic";
    
    @PostConstruct
    public void init() throws PulsarClientException {
        log.info("初始化Pulsar客户端，服务地址: {}", serviceUrl);
        
        client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .build();
        
        producer = client.newProducer()
                .topic(String.format("persistent://%s/%s/%s", tenant, namespace, TEST_TOPIC))
                .producerName("mq-study-producer")
                .create();
        
        log.info("Pulsar生产者初始化完成");
    }
    
    @PreDestroy
    public void destroy() throws PulsarClientException {
        if (producer != null) {
            producer.close();
        }
        if (client != null) {
            client.close();
        }
        log.info("Pulsar客户端已关闭");
    }
    
    /**
     * 发送同步消息
     */
    public void sendSyncMessage(String topic, Message message) {
        try {
            log.info("发送同步消息到主题: {}, 消息: {}", topic, message);
            
            String messageBody = objectMapper.writeValueAsString(message);
            String fullTopic = String.format("persistent://%s/%s/%s", tenant, namespace, topic);
            
            // 创建临时生产者
            try (Producer<String> tempProducer = client.newProducer()
                    .topic(fullTopic)
                    .producerName("sync-producer")
                    .create()) {
                
                tempProducer.send(messageBody);
                log.info("同步消息发送成功 - Topic: {}", topic);
            }
            
        } catch (JsonProcessingException e) {
            log.error("消息序列化失败: {}", e.getMessage(), e);
            throw new RuntimeException("消息序列化失败", e);
        } catch (PulsarClientException e) {
            log.error("发送同步消息失败: {}", e.getMessage(), e);
            throw new RuntimeException("发送同步消息失败", e);
        }
    }
    
    /**
     * 发送异步消息
     */
    public CompletableFuture<Void> sendAsyncMessage(String topic, Message message) {
        try {
            log.info("发送异步消息到主题: {}, 消息: {}", topic, message);
            
            String messageBody = objectMapper.writeValueAsString(message);
            String fullTopic = String.format("persistent://%s/%s/%s", tenant, namespace, topic);
            
            // 创建临时生产者
            Producer<String> tempProducer = client.newProducer()
                    .topic(fullTopic)
                    .producerName("async-producer")
                    .create();
            
            CompletableFuture<Void> future = new CompletableFuture<>();
            
            tempProducer.sendAsync(messageBody)
                    .thenAccept(msgId -> {
                        log.info("异步消息发送成功 - Topic: {}, MessageId: {}", topic, msgId);
                        try {
                            tempProducer.close();
                        } catch (PulsarClientException e) {
                            log.error("关闭生产者失败: {}", e.getMessage());
                        }
                        future.complete(null);
                    })
                    .exceptionally(throwable -> {
                        log.error("异步消息发送失败 - Topic: {}, Error: {}", topic, throwable.getMessage());
                        try {
                            tempProducer.close();
                        } catch (PulsarClientException e) {
                            log.error("关闭生产者失败: {}", e.getMessage());
                        }
                        future.completeExceptionally(throwable);
                        return null;
                    });
            
            return future;
            
        } catch (JsonProcessingException e) {
            log.error("消息序列化失败: {}", e.getMessage(), e);
            throw new RuntimeException("消息序列化失败", e);
        } catch (PulsarClientException e) {
            log.error("创建异步生产者失败: {}", e.getMessage(), e);
            throw new RuntimeException("创建异步生产者失败", e);
        }
    }
    
    /**
     * 发送批量消息
     */
    public void sendBatchMessages(String topic, int count) {
        log.info("发送批量消息到主题: {}, 数量: {}", topic, count);
        
        for (int i = 1; i <= count; i++) {
            Message message = Message.builder()
                    .messageId("batch-" + i)
                    .topic(topic)
                    .content("批量消息内容 " + i)
                    .messageType("BATCH")
                    .key("batch-key-" + i)
                    .build();
            
            try {
                sendSyncMessage(topic, message);
            } catch (Exception e) {
                log.error("发送批量消息失败 - 序号: {}, Error: {}", i, e.getMessage());
            }
        }
    }
    
    /**
     * 发送测试消息
     */
    public void sendTestMessage(Message message) {
        sendSyncMessage(TEST_TOPIC, message);
    }
    
    /**
     * 发送订单消息
     */
    public void sendOrderMessage(Message message) {
        sendSyncMessage(ORDER_TOPIC, message);
    }
    
    /**
     * 发送用户消息
     */
    public void sendUserMessage(Message message) {
        sendSyncMessage(USER_TOPIC, message);
    }
    
    /**
     * 发送流式消息
     */
    public void sendStreamMessage(Message message) {
        sendSyncMessage(STREAM_TOPIC, message);
    }
} 